package cn.gov.cus.rmftquartz.thread;

import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 子线程里也可以再开线程。
 * @author wxming.
 * @date 2018/1/30.
 */
@Component
public class ThreadDemo {
    private static final int POOL_SIZE=10;

    /*
    对于多个线程操作结果放在一个list中的需求，可直接把当前类里的一个线程安全的队列Queue类传到线程里，
    让线程操作同一个变量对象往里“.offer”放值，最终使多个线程中产生的内容在聚在一个list中，而不需要专门去返回再放，
    这样可以很方便的同时处理多个平行的数据而放在不同的队列Queue中（传多个Queue变量对象进去线程）。
    使用线程安全队列类（自己日志项目里用的那个）：Queue<EntryHead> userQueue=new ConcurrentLinkedDeque<>();或者ConcurrentLinkedQueue。
    ConcurrentLinkedDeque是两侧都可以进出的。而ConcurrentLinkedQueue只能从一侧进另一侧出。
    ConcurrentLinkedDeque是对ConcurrentLinkedQueue的扩展。别的特点一样用，都是非阻塞的，线程安全的，先进先出的。
     */

    /**
     * 调用不需要返回值的线程类。
     * @throws InterruptedException
     */
    public void runnableThread() throws InterruptedException {
        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(POOL_SIZE);

        String canshu1="aa";
        String canshu2="bb";
        for(int i=0;i<5;i++){
            //因为线程是new出来的，不归spring管理。所以里面不能@Autowired注入对象，线程里可通过构造函数传需要的参数或者spring管理的service或者dao类。
            RunnableThread runnableThread = new RunnableThread(canshu1,canshu2);

            //预防等待队列过载
            if(executor.getQueue().size()>POOL_SIZE){
                try{
                    executor.awaitTermination(100,TimeUnit.MILLISECONDS);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            //提交线程。
            executor.submit(runnableThread);
        }
        //结束线程池
        executor.shutdown();
        //等待子线程全部执行完毕（最大等待时间：1Day)
        executor.awaitTermination(1, TimeUnit.DAYS);
    }

    /**
     * 调用有返回值的线程类。
     * @throws InterruptedException
     */
    public void callBackThread() throws InterruptedException {
        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(POOL_SIZE);
        //接收线程返回的List<String>(也可以是别的任意类型）。Future里的返回类型可根据线程里的需求随意定。
        List<Future<List<String>>> listResultBean = new ArrayList<>();

        String canshu1="aa";
        String canshu2="bb";
        for(int i=0;i<5;i++){
            //因为此处线程是new出来的，不归spring管理。所以里面不能@Autowired注入对象，线程里可通过构造函数传需要的参数或者spring管理的service或者dao类。
            CallableThread callableThread = new CallableThread(canshu1,canshu2);

            //预防等待队列过载
            if(executor.getQueue().size()>POOL_SIZE){
                try{
                    executor.awaitTermination(100,TimeUnit.MILLISECONDS);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            //提交线程，并将线程返回的值放入listResultBean。
            listResultBean.add(executor.submit(callableThread));
        }


/*
下面两个“结束”“等待”方法必须都写。结束线程池shutdown()除了关闭线程池外，还会让子线程都走完才关闭。但并不会都走完才会往下走之后别的代码，而是往下走的同时走子线程。
等待awaitTermination方法会等待executor里的任务都走完，但并不只是子线程，还有别的自带的不知道什么任务。
1、如果只写结束shutdown，则之后方法中获得的从子线程来的数据会有问题。并不会等子线程都走完才执行之后的方法。
2、如果只写等待awaitTermination，则即使子线程都走完也还会等待很长时间，很长很长时间。
3、所以需要两者都写结合使用才行。此处shutdown负责关闭线程池中其他自带的任务，awaitTermination负责线程池中没有任务了再往下走下面的代码，以确保来自子线程的数据的正确性。
4、还有一个方法是executor.shutdownNow()，会立刻关闭线程池，而不会等待子线程走完，不符合需求一般不用。
*/
        //结束线程池
        executor.shutdown();
        //等待子线程全部执行完毕（最大等待时间：1Day)
        executor.awaitTermination(1, TimeUnit.DAYS);
        //解析线程里返回的数据。
        getListObjToSpark(listResultBean);
    }


    /**
     * 解析线程里返回的数据。
     * @param listResultBean
     * @return
     */
    private List<String> getListObjToSpark(List<Future<List<String>>> listResultBean){
        List<String> listObjSpark=new ArrayList<>();
        for (Future<List<String>> listObj:listResultBean){
            try{
                //listObj.get()是Future<>的方法，获得当前是Future内的对象。Future<>内可以放任何类型，跟线程里返回的类型一致就行。
                for(String x:listObj.get()){
                    listObjSpark.add(x);
                }
            }catch (InterruptedException e){

            }catch (ExecutionException e){

            }
        }
        return listObjSpark;
    }

    private List<Object> getListObjToSpark2(List<Future<Map<String,Object>>> listResultBean){
        List<Object> listObjSpark=new ArrayList<>();
        for (Future<Map<String,Object>> listObj:listResultBean){
            try{
                if(listObj!=null){//此if判断不需要。肯定不会是null。即使线程里出错，会把错误信息放进listObjSpark，但会因为类型不一致此处报错。如果线程里进行了trycatch将出错返回null，会将null放进listObjSpark，listObjSpark的size依然增加，只不过debug查看listObjSpark时放null的位置不显示，list的size数直接跳过了。
                    listObjSpark.add(listObj.get());
                }
            }catch (InterruptedException e){//trycatch要放在for循环里面，不要放在for循环外。否则一出错被循环的对象里的剩下的数据都不走了。

            }catch (ExecutionException e){

            }
        }
        return listObjSpark;
    }
}
